package realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

/**
 * @Create 2023/9/5 20:41
 */
public class DimBroadcastFunction extends BroadcastProcessFunction<JSONObject,JSONObject,JSONObject> {

    private MapStateDescriptor<String,JSONObject> descriptor;

    private Map<String,JSONObject> configMap=new HashMap<>();

    public DimBroadcastFunction(MapStateDescriptor<String, JSONObject> descriptor) {
        this.descriptor = descriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:13306/edu_config?"+
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        PreparedStatement prepareStatement = conn.prepareStatement("select * from edu_config.table_process");
        ResultSet resultSet = prepareStatement.executeQuery();

        //处理结果集
        ResultSetMetaData metaData = resultSet.getMetaData();
        while (resultSet.next()){
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object object = resultSet.getObject(i);
                jsonObject.put(columnName,object);
            }
            configMap.put(jsonObject.getString("source_table"),jsonObject);
        }

        resultSet.close();
        prepareStatement.close();
        conn.close();
    }


    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, JSONObject, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, JSONObject> broadcastState = ctx.getBroadcastState(descriptor);

        String table = value.getString("table");
        JSONObject jsonObject=null;
        if ((jsonObject=broadcastState.get(table))!=null||(jsonObject=configMap.get(table))!=null){
            JSONObject dataJson = value.getJSONObject("data");
            String sinkColumns = jsonObject.getString("sink_columns");
            filterColumn(dataJson,sinkColumns);

            //补充操作类型
            dataJson.put("type",value.getString("type"));
            //补充配置表信息
            dataJson.put("table_process",jsonObject);

            out.collect(dataJson);
        }
    }

    /**
     * 过滤掉不需要向下游传递的属性
     * @param dataJson
     * @param sinkColumns
     */
    private void filterColumn(JSONObject dataJson, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);

        Set<Map.Entry<String, Object>> entrySet = dataJson.entrySet();
        entrySet.removeIf(new Predicate<Map.Entry<String, Object>>() {
            @Override
            public boolean test(Map.Entry<String, Object> stringObjectEntry) {
                return !columnList.contains(stringObjectEntry.getKey());
            }
        });
    }

    /**
     * 处理处理广播流,通过op判断是否删除,如果更新都更新删除都删除,将sourcetable作为key,维度jsonObj作为value保存
     * @param value The stream element.
     * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
     *     current processing/event time and updating the broadcast state. The context is only valid
     *     during the invocation of this method, do not store it.
     * @param out The collector to emit resulting elements to
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(JSONObject value, BroadcastProcessFunction<JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, JSONObject> broadcastState = ctx.getBroadcastState(descriptor);
        String sourceTable = value.getString("source_table");
        String op = value.getString("op");
        if (op.equals("d")){
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        }else {
            broadcastState.put(sourceTable,value);
            configMap.put(sourceTable,value);
        }
    }
}
